package com.example.mq.factory;

import com.example.mq.annotation.MQListener;
import com.example.mq.config.SysMQConfig;
import com.example.mq.dto.MQListenerDto;
import com.example.mq.listener.SysMQMessageListener;
import com.example.mq.po.SysMqReceiveLog;
import com.example.mq.service.SysMqReceiveLogService;
import com.example.server.service.SysServerService;
import com.example.sync.handler.impl.SysMQMessageRetryHandler;
import com.example.sync.handler.impl.wrap.SysMQRetryMessage;
import lombok.Data;
import lombok.SneakyThrows;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import org.springframework.jms.connection.CachingConnectionFactory;
import org.springframework.util.StringUtils;

import javax.jms.*;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/*
 * @className: SysRabbitMQFactory
 * @author: chentao
 * @date: 2025/3/25 下午2:19
 * @Version: 1.0
 * @description:
 */
@Data
public class SysActiveMQFactory extends SysAbstractMQFactory {

    Logger logger = LoggerFactory.getLogger(getClass());

    private volatile boolean running = true;

    private SysMQConfig sysMQConfig;

    private ActiveMQConnectionFactory activeMQConnectionFactory;

    private CachingConnectionFactory connectionFactory;

    private Connection connection;

    private ExecutorService executorService;

    private SysServerService sysServerService;

    private ApplicationContext applicationContext;

    private SysMqReceiveLogService sysMqReceiveLogService;

    private SysMQMessageRetryHandler sysMQMessageRetryHandler;

    @Override
    public void init(SysMQConfig sysMQConfig, ApplicationContext applicationContext, SysServerService sysServerService) {
        try {
            this.executorService = Executors.newCachedThreadPool();
            this.sysMQConfig = sysMQConfig;
            this.applicationContext = applicationContext;
            this.sysServerService = sysServerService;
            this.sysMQMessageRetryHandler = applicationContext.getBean(SysMQMessageRetryHandler.class);
            this.sysMqReceiveLogService = applicationContext.getBean(SysMqReceiveLogService.class);
            this.activeMQConnectionFactory = new ActiveMQConnectionFactory(sysMQConfig.getUsername(), sysMQConfig.getPassword(), "tcp://" + sysMQConfig.getHost() + ":" + sysMQConfig.getPort());
            this.connectionFactory = new CachingConnectionFactory(activeMQConnectionFactory);
            this.connection = this.connectionFactory.createConnection();
            this.connection.start();
        } catch (JMSException e) {
            logger.warn("ActiveMQ服务 启动失败：｛｝", e);
            if (this.connection != null) {
                try {
                    this.connection.close();
                } catch (JMSException ex) {
                    logger.warn("关闭connection异常: " + ex.getMessage());
                }
            }
        }
    }

    @Override
    public void preValidate(Map<String, MQListenerDto> mqListenerDtoMap) throws Exception {
        for (MQListenerDto mqListenerDto : mqListenerDtoMap.values()) {
            MQListener mqListener = mqListenerDto.getMqListener();
            if (StringUtils.isEmpty(mqListener.queue())) {
                throw new Exception("bean: " + mqListenerDto.getBeanName() + " 对应queue 不可以为空.");
            }
        }
    }

    @Override
    public void start(Map<String, MQListenerDto> mqListenerDtoMap) {
        for (String name : mqListenerDtoMap.keySet()) {
            MQListenerDto mqListenerDto = mqListenerDtoMap.get(name);
            executorService.execute(new ActiveMQMessageListenerThread(mqListenerDto));
        }
    }


    @Override
    public void stop() throws Exception {
        if (this.connection != null) {
            try {
                this.connection.close();
            } catch (JMSException e) {
                logger.warn("close connection error", e);
            }
        }
        this.executorService.shutdownNow();
    }

    //mq消息监听线程
    class ActiveMQMessageListenerThread implements Runnable {

        private MQListenerDto mqListenerDto;

        public ActiveMQMessageListenerThread(MQListenerDto mqListenerDto) {
            this.mqListenerDto = mqListenerDto;
        }

        @Override
        public void run() {
            MessageConsumer messageConsumer = null;
            Session session = null;
            try {
                MQListener mqListener = mqListenerDto.getMqListener();
                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                if (mqListenerDto.getMqListener().isTopic()) {
                    //主题信息
                    Topic topic = session.createTopic(mqListener.queue());
                    messageConsumer = session.createConsumer(topic);
                } else {
                    //队列消息
                    Queue queue = session.createQueue(mqListener.queue());
                    messageConsumer = session.createConsumer(queue);
                }
                while (running) {
                    try {
                        Message message = messageConsumer.receive();
                        if (message instanceof TextMessage) {
                            SysMQMessageListener sysMQMessageListener = (SysMQMessageListener) applicationContext.getBean(mqListenerDto.getBeanName());
                            executorService.execute(new ActiveMQMessageReceiveThread(mqListenerDto, sysMQMessageListener, ((TextMessage) message).getText()));
                        }
                    } catch (Exception e) {
                        if (!(e instanceof InterruptedException || e instanceof JMSException)) {
                            logger.warn("监听ActiveMQ消息线程失败: " + e.getMessage());
                        }
                    }
                }
            } catch (Exception e) {
                logger.warn("启动监听ActiveMQ消息线程失败: " + e.getMessage());
            } finally {
                if (messageConsumer != null) {
                    try {
                        messageConsumer.close();
                    } catch (JMSException e) {
                        logger.warn("关闭消息接受出错: " + e.getMessage());
                    }
                }
                if (session != null) {
                    try {
                        session.close();
                    } catch (JMSException e) {
                        logger.warn("关闭session异常: " + e.getMessage());
                    }
                }
            }
        }
    }

    //MQ消息的处理
    class ActiveMQMessageReceiveThread implements Runnable {

        private MQListenerDto mqListenerDto;

        private SysMQMessageListener sysMQMessageListener;

        private String message;

        public ActiveMQMessageReceiveThread(MQListenerDto mqListenerDto, SysMQMessageListener sysMQMessageListener, String message) {
            this.mqListenerDto = mqListenerDto;
            this.sysMQMessageListener = sysMQMessageListener;
            this.message = message;
        }

        @SneakyThrows
        @Override
        public void run() {
            SysMqReceiveLog sysMqReceiveLog = null;
            try {
                sysMqReceiveLog = new SysMqReceiveLog(SysActiveMQFactory.class.getName(), mqListenerDto.getMqListener().queue(), mqListenerDto.getBeanName(), sysServerService.getCurrentServerId(), message);
                sysMQMessageListener.onMessage(message);
            } catch (Exception e) {
                sysMqReceiveLog.setStatus(false);
                SysMQRetryMessage sysMQRetryMessage = new SysMQRetryMessage(mqListenerDto.getMqListener().queue(), mqListenerDto.getBeanName(), message);
                sysMQMessageRetryHandler.publish(SysActiveMQFactory.class.getName(), 0L, "mq.message.retry", sysMQRetryMessage.getClass().getName(), sysMQRetryMessage);
                logger.warn("mq bean :" + sysMQMessageListener.getClass().getName() + " 消费消息失败: " + e.getMessage());
            } finally {
                sysMqReceiveLog.recalculateFields();
                sysMqReceiveLogService.save(sysMqReceiveLog);
            }
        }
    }

    @Override
    public void publish(String queue, boolean isTopic, String content) throws Exception {
        super.publish(queue, isTopic, content);
        Session session = null;
        MessageProducer messageProducer = null;
        try {
            session = this.connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
            if (isTopic) {
                TextMessage textMessage = session.createTextMessage(content);
                Topic topic = session.createTopic(queue);
                messageProducer = session.createProducer(topic);
                messageProducer.send(textMessage);
            } else {
                TextMessage textMessage = session.createTextMessage(content);
                messageProducer = session.createProducer(session.createQueue(queue));
                messageProducer.send(textMessage);
            }
        } catch (Exception e) {
            logger.warn("MQ推送消息失败: " + e.getMessage());
            throw e;
        } finally {
            if (messageProducer != null) {
                messageProducer.close();
            }
            if (session != null) {
                session.close();
            }
        }
    }
}
